package com.btcode.netty.server.handler;

import com.btcode.common.MyLog;
import com.btcode.log.ILog;
import com.btcode.netty.server.session.IoSession;
import com.btcode.netty.server.session.SessionManager;
import com.btcode.netty.server.session.util.ChannelUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.SingleThreadEventExecutor;
import sun.management.ManagementFactoryHelper;

import java.lang.management.BufferPoolMXBean;
import java.lang.reflect.Field;
import java.util.Queue;

/**
 * 2.用于会话管理（可选，需按要求实现抽象方法），比如连接管理、冲突检测（需覆盖shouldCheckConflictAndUseNewConn）、消息分发SessionManager（sendMsg2ClientByChannel、sendMsg2ClientById）
 * 4.SimpleChannelInboundHandler会对channelRead0后的ByteBuf自动回收
 */
public final class HrtSessionCtrl extends SimpleChannelInboundHandler {
    protected final ILog log = MyLog.getInstance().getLogger(HrtSessionCtrl.class);
    private AbstractHrtDecoder ahd;
    private String mn = null;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (mn == null) {
            mn = ahd.getMnFromChannel(ctx, (ByteBuf) msg);
            //首次接入时，判断是否已经存在mn相同，channel不同的会话
            if (mn != null) {
                if (ahd.shouldCheckConflictAndUseNewConn()[0]) {
                    IoSession session = SessionManager.INSTANCE.getSessionById(mn);
                    if (session != null && !ctx.channel().equals(session.getChannel())) {
                        log.error(String.format("已经存在mn均为%s[%s]，channel不同的会话%s，准备关闭%s连接",
                            mn, ChannelUtil.getIp(ctx.channel()), session, ahd.shouldCheckConflictAndUseNewConn()[1] ? "旧" : "新"));
                        if (ahd.shouldCheckConflictAndUseNewConn()[1]) {
                            SessionManager.INSTANCE.unRegisterUserContext(mn);
                        }
                        else {
                            ReferenceCountUtil.release(msg);
                            SessionManager.INSTANCE.unRegisterUserContext(ctx.channel());
                            return;
                        }
                    }
                }
                SessionManager.INSTANCE.registerSession(mn, ctx.channel());
            }
        }
        ctx.fireChannelRead(msg);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        try {//添加netty监控
            monitorPendingTaskCount(ctx);
            monitorQueueFirstTask(ctx);
            monitorOutboundBufSize(ctx);
            monitorDirect();
        }
        catch (Exception e) {
            log.error(ChannelUtil.getIp(ctx.channel()), e);
        }
    }

    @Override
    public final void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        SessionManager.INSTANCE.unRegisterUserContext(ctx.channel());
    }

    private void monitorDirect() {
        BufferPoolMXBean mxBean = ManagementFactoryHelper.getBufferPoolMXBeans().get(0);
        log.info(String.format("hasCleaner的DirectBuffer的数量=%s，堆外内存占用大小=%s", mxBean.getCount(), mxBean.getMemoryUsed()));

    }

    /**
     * 监控任务队列堆积任务数，任务队列中的任务包括io读写任务，业务程序提交任务
     */
    private void monitorPendingTaskCount(ChannelHandlerContext ctx) {
        int totalPendingSize = 0;
        for (EventExecutor eventExecutor : ctx.executor().parent()) {
            SingleThreadEventExecutor executor = (SingleThreadEventExecutor) eventExecutor;
            // 注意，Netty4.1.29以下版本本pendingTasks()方法存在bug，导致线程阻塞问题
            // 参考 https://github.com/netty/netty/issues/8196
            totalPendingSize += executor.pendingTasks();
        }
        log.info("任务队列中总任务数 = " + totalPendingSize);
    }

    /**
     * 监控各个堆积的任务队列中第一个任务的类信息
     */
    private void monitorQueueFirstTask(ChannelHandlerContext ctx) throws NoSuchFieldException, IllegalAccessException {
        Field singleThreadField = SingleThreadEventExecutor.class.getDeclaredField("taskQueue");
        singleThreadField.setAccessible(true);
        for (EventExecutor eventExecutor : ctx.executor().parent()) {
            SingleThreadEventExecutor executor = (SingleThreadEventExecutor) eventExecutor;
            @SuppressWarnings("unchecked")
            Runnable task = ((Queue<Runnable>) singleThreadField.get(executor)).peek();
            if (null != task) {
                log.info("任务队列中第一个任务信息：" + task.getClass().getName());
            }
        }
    }

    /**
     * 监控出站消息的队列积压的byteBuf大小
     */
    private void monitorOutboundBufSize(ChannelHandlerContext ctx) {
        long outBoundBufSize = ((NioSocketChannel) ctx.channel()).unsafe().outboundBuffer().totalPendingWriteBytes();
        log.info("出站消息队列中积压的buf大小" + outBoundBufSize);
    }

    /**
     * 由于解码器中decode和channelRead本身冲突，会话管理只能转移
     * 解码器中会话管理相关接口为了避免转移，通过引用传递
     *
     * @param ahd 解码器
     * @return
     */
    public HrtSessionCtrl setDecoderRef(AbstractHrtDecoder ahd) {
        this.ahd = ahd;
        return this;
    }
}
